package com.cskaoyan.mall.search.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.search.service.SearchService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @Author Pudding
 * @Date 2024/7/16 17:50
 */
@Component
public class MqOnsaleConsumer {

    @Autowired
    SearchService searchService;

    DefaultMQPushConsumer defaultMQPushConsumer;

    @PostConstruct
    public void init() throws MQClientException {

        //1. 创建消息消费者对象
        defaultMQPushConsumer = new DefaultMQPushConsumer("onsale-group");

        //2. 设置注册中心
        defaultMQPushConsumer.setNamesrvAddr("47.96.11.99:9876");

        //3. 订阅主题
        defaultMQPushConsumer.subscribe(MqTopicConst.PRODUCT_ONSALE_TOPIC, "*");

        //4. 设置消息监听器
        defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext content) {
                try {
                    //4.1 获取消息对象
                    MessageExt messageExt = msgs.get(0);
                    //4.2 获取消息内容
                    byte[] body = messageExt.getBody();
                    String messageContent = new String(body);

                    Long skuId = JSON.parseObject(messageContent, Long.class);

                    //执行业务逻辑：上架
                    searchService.upperGoods(skuId);

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        //5. 启动
        defaultMQPushConsumer.start();

    }
}
